package com.tjbklx1.zkclient.mastersel;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkException;
import org.I0Itec.zkclient.exception.ZkInterruptedException;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.zookeeper.CreateMode;

import com.tjbklx1.constant.Host;

/**
 * 主工作类
 * 
 * @author Administrator
 * 
 */
public class WorkServer {

	// 服务器状态
	private volatile boolean running = false;
	// 客户端
	private ZkClient zkClient;
	// 监听MASTER的删除事件
	private IZkDataListener dataListener;
	// 当前服务器的基本信息
	private WorkServerData serverData;
	// master的基本信息
	private WorkServerData masterData;

	//调度器 应对 网络抖动
	private ScheduledExecutorService delayExector = Executors
			.newScheduledThreadPool(1);
	private int delayTime = 5;

	public WorkServer(WorkServerData rd) {
		this.serverData = rd;
		this.dataListener = new IZkDataListener() {
			
			/**
			 * 节点删除事件
			 */
			public void handleDataDeleted(String dataPath) throws Exception {
				// takeMaster();

				//应对网络抖动,上一轮的master=server 先去争抢master,否则延迟5秒争抢
				if (masterData != null
						&& masterData.getName().equals(serverData.getName())) {
					takeMaster();

				} else {
					delayExector.schedule(new Runnable() {
						public void run() {
							takeMaster();
						}
					}, delayTime, TimeUnit.SECONDS);

				}

			}

			public void handleDataChange(String dataPath, Object data)
					throws Exception {

			}
		};
	}

	public ZkClient getZkClient() {
		return zkClient;
	}

	public void setZkClient(ZkClient zkClient) {
		this.zkClient = zkClient;
	}

	/**
	 * 启动
	 * @throws Exception
	 */
	public void start() throws Exception {
		//检查服务器是否处于运行状态
		if (running) {
			throw new Exception("server has startup...");
		}
		running = true;
		//订阅master的事件
		zkClient.subscribeDataChanges(Host.MASTER_PATH, dataListener);
		//争抢MATER权利
		takeMaster();

	}

	/**
	 * 停止服务
	 * @throws Exception
	 */
	public void stop() throws Exception {
		//检查服务器是否处于停止状态
		if (!running) {
			throw new Exception("server has stoped");
		}
		running = false;

		delayExector.shutdown();
		//取消master节点的事件订阅
		zkClient.unsubscribeDataChanges(Host.MASTER_PATH, dataListener);
		//释放master权利
		releaseMaster();

	}

	/**
	 * 争抢MATER权利
	 */
	private void takeMaster() {
		if (!running)
			return;

		try {
			//尝试创建master节点
			zkClient.create(Host.MASTER_PATH, serverData, CreateMode.EPHEMERAL);
			//serverdata 赋值给 masterdata
			masterData = serverData;
			System.out.println(serverData.getName() + " is master");
			
			//演示 : 每隔5秒释放master权利
			delayExector.schedule(new Runnable() {
				public void run() {
					if (checkMaster()) {
						releaseMaster();
					}
				}
			}, 5, TimeUnit.SECONDS);

		} catch (ZkNodeExistsException e) {
			//读取当前节点的信息 ,放入masterdata中
			WorkServerData runningData = zkClient.readData(Host.MASTER_PATH, true);
			//如果没有data ,说明master宕机,再次takemaster
			if (runningData == null) {
				takeMaster();
			} else {
				masterData = runningData;
			}
		} catch (Exception e) {
			// ignore;
			e.printStackTrace();
		}

	}

	/**
	 * 释放MASTER权利
	 */
	private void releaseMaster() {
		if (checkMaster()) {
			zkClient.delete(Host.MASTER_PATH);

		}

	}

	/**
	 * 检测自己是不是MASTER
	 * 
	 * @return
	 */
	private boolean checkMaster() {
		try {
			WorkServerData eventData = zkClient.readData(Host.MASTER_PATH);
			masterData = eventData;
			if (masterData.getName().equals(serverData.getName())) {
				return true;
			}
			return false;
		} catch (ZkNoNodeException e) {
			//节点不存在
			return false;
		} catch (ZkInterruptedException e) {
			//处理中断异常 是重试
			return checkMaster();
		} catch (ZkException e) {
			return false;
		}
	}

}
